package day3

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession

object SparkSQL {
  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "D:\\devtools\\hadoop")
    Logger.getLogger("org").setLevel(Level.OFF)

    /*
     * SparkSQL：结构化、半结构化数据
     * Hive、JDBC
     * 需要处理结构化数据的时候，这个时候SparkSQL会更管用
     * 如果说SparkSQL需要对接Hive，一定要Hive的配置hive-site.xml放置在Spark配置文件
     * SparkSQL可以独立运行
     */
    // getOrCreate()：检查会话是否存在，如果存在则正常返回，不存在则创建
    val session = SparkSession.builder()
      .appName("SparkSQL-test")
      .master("local")
      .getOrCreate()
    val sc = session.sparkContext
    //使用RDD的方式将自定义数据导入
    val value = sc.parallelize(List(
      (1, "zhangsan", "math", 60), (2, "lisi", "chinese", 140),
      (3, "wangwu", "math", 120), (4, "zhaoliu", "chinese", 100)
    ))
    //使用RDD的方式将文件导入
    //    sc.textFile("D:\\data\\files\\consume.log")
    //使用session方式将文件导入
    //    session.read.textFile("D:\\data\\files\\consume.log")
    /*
     * rdd会默认以\n来进行一组数据的读取，并且每个切割出来的数据都是String类型
     * session方式会默认以\n来进行一组数据的读取，根据每个数据的类型都定义为一个Row类型
     * * Row类型：
     * * * Row类型独属于Dataframe类型，是一个对象，本质上来说是一个定长的字段数组
     * * * 获得到Row类型的值：getter()方法
     * * * 通过值进行基础类型判定：getType()方法
     * DataFrame和DataSet
     * * 现在的版本中已经变得非常模糊，因为都可以使用SQL语句进行调用
     * * DataSet是强类型定义，必须返回一个元组类型
     * * DataFrame是Row类型，从逻辑上可以大致认为Row类型也是一个元组
     * * DataFrame实际上是DataSet的特殊形态
     */
    //使用RDD的方式进行处理
    //1 全部输出
    //    value.foreach(println)
    //2 看数学成绩
    //    value.filter(_._3 == "math").foreach(println)
    //2 看成绩大于100
    //    value.filter(_._4 >= 100).foreach(println)

    //把RDD转换成为Dataframe
    //在Scala当中隐式转换对于基础语法能够很直接的转换，在SparkSQL不行
    import session.implicits._
    //dataFrame需要定义字头
    val value_DF = value.toDF("id", "name", "course", "grade")
    //dataSet不需要定义字头
    val value_DS = value.toDS()

    //1、将value_DF和value_DS输出
    //    value_DF.show()
    //    value_DS.show()
    //2、只看数学成绩，输出姓名与成绩
    //不是一个实际生成的数据库，临时的会话不会在物理层面创建
    value_DF.createOrReplaceTempView("value_DF")
    value_DS.createOrReplaceTempView("value_DS")

    //使用SparkSQL实现
    //    session.sql("select name,course from value_DF where grade > 60").show()
    //    session.sql("select _2,_3 from value_DS where _4 > 60").show()
    //使用RDD实现
    //    value.filter(_._4 > 60).foreach(i => {
    //      println(i._2 + "\t" + i._3)
    //    })

    //DataFrame、DataSet转换为RDD
    //getAs
    value_DF.rdd.map(_.getAs[String]("name")).foreach(println)
    value_DS.rdd.map(_._2).foreach(println)
    value_DF.rdd.map(i =>
      ( i.getAs("name"),
        i.getAs("course"),
        i.getAs("grade")
        )
    ).foreach(println)
    //关闭SparkSQL
    session.close()
  }
}
